package docsim;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.HashMap;
import java.util.StringTokenizer;

import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.util.StringUtils;

public class Map2 extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> {

	private static final String SLASH = "/";
	private static final int ZERO = 0;
	private static final int ONE = 1;
	private static String colon = ":";
	private static String space = " ";
	private static int docid;
	private static Text docidCount;
	private Text wordToPut = new Text();
	public String inputFile;
	
	public  HashMap<Integer,String> idDocMap = new HashMap<Integer,String>(); 
    public  HashMap<String,Integer> docIdMap = new HashMap<String,Integer>();

	@Override
	public void map(LongWritable key, Text value,
			OutputCollector<Text, Text> collector, Reporter reporter)
			throws IOException {
		//long b = System.currentTimeMillis();
		//System.out.println("Map2!");
 		//System.out.println(docIdMap.size());
 		inputFile = inputFile.substring(inputFile.lastIndexOf(SLASH)+1);
 		//System.out.println(inputFile);
 		Integer i = docIdMap.get(inputFile);
		docid = i;
		
		//long p1 = System.currentTimeMillis();
		String content = Preprocessor.preprocessText(value.toString());
		//long p2 = System.currentTimeMillis();
        //System.out.println("pre-process time"+":"+(p2-p1));
        //String content = value.toString();
		String[] words = content.split(space);
		HashMap<String, Integer> wordCount = new HashMap<String, Integer>();
		for (String word : words) {
			if (wordCount.containsKey(word)) {
				wordCount.put(word, wordCount.get(word) + ONE);
			} else {
				wordCount.put(word, ONE);
			}
		}
		for (String word : wordCount.keySet()) {
			if(word.length()!=ZERO)
			{
			wordToPut.set(word);
			docidCount = new Text(docid+colon+wordCount.get(word));
			// docidCount.set(docid.get()+":"+wordCount.get(word));
			collector.collect(wordToPut, docidCount);
			}
		}
		//long a = System.currentTimeMillis();
        //double id = Math.random();
        //System.out.println("Map2 Time:"+id+":"+(a-b));
	}
	
 	private void loadDocIdMap(Path docIdsFile) {
        try {
          BufferedReader fis = new BufferedReader(new FileReader(docIdsFile.toString()));
          String pattern = null;
          Integer key;
          String value;
          while ((pattern = fis.readLine()) != null) {
       	    StringTokenizer st1 = new StringTokenizer(pattern);
       	    key = Integer.parseInt(st1.nextToken());
       	    value = st1.nextToken();
       	    idDocMap.put(key, value);
       	    docIdMap.put(value, key);
          }
        } catch (IOException ioe) {
          System.err.println("Caught exception while parsing the cached file '" + idDocMap + "' : " + StringUtils.stringifyException(ioe));
        }
      }
 	
   @Override
   public void configure(JobConf job) {
	   //long b = System.currentTimeMillis();
       inputFile = job.get("map.input.file");
       
       if (job.getBoolean("docsim.docids.map", true)) {
           Path[] docidMapFile = null;
           try {
             docidMapFile = DistributedCache.getLocalCacheFiles(job);
           } catch (IOException ioe) {
             System.err.println("Caught exception while getting cached files: " + StringUtils.stringifyException(ioe));
           }
           for (Path docidFile : docidMapFile) {
             //if(docidFile.getName().matches("docid_map.txt"))
            	loadDocIdMap(docidFile);
           }
         }
       //long a = System.currentTimeMillis();
       //double id = Math.random();
       //System.out.println("Config2 Time:"+id+":"+(a-b));
     }
}